package cn.xuexiyuan.flinkstudy.watermark;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Description: 基与事件时间的窗口计算 + watermark 解决一定程度数据乱序/延迟到达的问题
 *  并【侧道输出】使用 outputTag + allowedLateness 来解决数据丢失问题 （解决迟到严重的数据问题）
 *
 * @Author 左龙龙
 * @Date 21-3-25
 * @Version 1.0
 **/
public class WatermarkDemo03 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            private boolean running = true;
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while(running){
                    Order order = DataFactory.createRandomOrder(3, 5);
                    System.out.println("->" + order);
                    ctx.collect(order);

                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        // 2.transformation
        // 在最新版本中默认的就是 EventTime
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        SingleOutputStreamOperator<Order> orderDSWithWatermark = orderDS.assignTimestampsAndWatermarks(
                // 设置 maxOutOfOrderness 最大无序度/最大允许的延迟/乱序时间
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        // 指定事件时间列
                        .withTimestampAssigner((order, timestamp) -> order.getOrder_time())
        );

        // 准备用来存放迟到严重的数据
        OutputTag<Order> seriousLateOutputTag = new OutputTag<>("seriousLate", TypeInformation.of(Order.class));

        SingleOutputStreamOperator<Order> result = orderDSWithWatermark
                .keyBy(Order::getUser_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 允许数据迟到/延迟的时间
                .allowedLateness(Time.seconds(3))
                // 指定迟到严重数据的存储位置
                .sideOutputLateData(seriousLateOutputTag)
                .sum("payment_money");

        // 获取单独收集到迟到严重的数据
        DataStream<Order> sideOutput = result.getSideOutput(seriousLateOutputTag);

        // 3.sink
        result.print("正常数据");
        sideOutput.print("迟到严重数据");


        // 4.excute
        env.execute();

    }

}
